package com.baoteng.rbspy.config;

import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.interceptor.CacheErrorHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;

import redis.clients.jedis.JedisPoolConfig;

@Configuration
//@EnableRedisHttpSession(maxInactiveIntervalInSeconds = 20 * 60,redisNamespace="sales",redisFlushMode=RedisFlushMode.IMMEDIATE)
public class RedisConfiguration extends CachingConfigurerSupport {

	private final Logger logger = LoggerFactory.getLogger(RedisConfiguration.class);

	@Value("${spring.redis.host}")
	private String host;

	@Value("${spring.redis.port}")
	private Integer port;

	@Value("${spring.redis.pool.max-active}")
	private Integer maxActive;
	
	@Value("${spring.redis.pool.max-idle}")
	private Integer maxIdle;
	
	@Value("${spring.redis.pool.max-wait}")
	private Integer maxWait;
	
	@Value("${spring.redis.pool.min-idle}")
	private Integer minIdle;

	@Value("${redis.enableTransactionSupport}")
	private boolean enableTransactionSupport;

	@Value("${spring.redis.database}")
	private Integer databse;

	@Value("${redis.sub.receiverTopic}")
	private String receiverTopic;

	@Value("${spring.redis.password}")
	private String password;
	
	@Autowired
    private MessageListenerAdapter receiverListenerAdapter;
    
//    @Autowired
//    private MessageListenerAdapter senderListenerAdapter;
    
	@Bean
	RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
		//TODO 修改RedisMessageListenerContainer缺省Executor subscriptionExecutor 。提高性能
		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
		container.setConnectionFactory(connectionFactory);
		container.addMessageListener(receiverListenerAdapter, new PatternTopic(receiverTopic));
		container.afterPropertiesSet();
		return container;
	}

	@Bean
	MessageListenerAdapter receiverListenerAdapter(RedisMessageReceiver receiverTopic) {
		return new MessageListenerAdapter(receiverTopic, "receiveMessage");
	}
	
	@Bean
	RedisMessageReceiver receiverTopic(CountDownLatch latch) {
		return new RedisMessageReceiver(latch);
	}

	@Bean
	CountDownLatch latch() {
		return new CountDownLatch(1);
	}

	@Bean
	public RedisConnectionFactory connectionFactory() {
		JedisPoolConfig poolConfig = new JedisPoolConfig();
		poolConfig.setMaxTotal(maxActive);
		poolConfig.setMaxWaitMillis(maxWait);
		poolConfig.setMaxIdle(maxIdle);
		poolConfig.setMinIdle(minIdle);
		poolConfig.setTimeBetweenEvictionRunsMillis(0);
		poolConfig.setTestWhileIdle(false);
//		poolConfig.setBlockWhenExhausted(false);
		poolConfig.setTestOnBorrow(true);
		JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(poolConfig);
		jedisConnectionFactory.setHostName(host);
		jedisConnectionFactory.setPort(port);
		jedisConnectionFactory.setPassword(password);
		jedisConnectionFactory.setDatabase(databse);
		jedisConnectionFactory.afterPropertiesSet();
		// RedisSentinelConfiguration sentinelConfig = new
		// RedisSentinelConfiguration() .master("r_master").sentinel(host,
		// port).sentinel(host, port);
		logger.info("jedis pool id building.");
		return jedisConnectionFactory;
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	private void setDataSerializer(RedisTemplate template) {
		Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(
				Object.class);
		ObjectMapper om = new ObjectMapper();
		om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
		om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
		om.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
		om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
		jackson2JsonRedisSerializer.setObjectMapper(om);
		template.setKeySerializer(template.getStringSerializer());
		template.setValueSerializer(jackson2JsonRedisSerializer);
		template.setHashValueSerializer(jackson2JsonRedisSerializer);
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	private void setMsgSerializer(RedisTemplate template) {
//		template.setKeySerializer(template.getStringSerializer());
		template.setValueSerializer(template.getStringSerializer());
		template.setHashValueSerializer(template.getStringSerializer());
	}

	@Bean(name = "jsonsRedisTemplate")
	public RedisTemplate<String,Object> jsonRedisTemplate() {
		RedisTemplate<String,Object> template = new RedisTemplate<String,Object>();
		template.setConnectionFactory(connectionFactory());
		setDataSerializer(template);
		template.setEnableTransactionSupport(false);
		template.afterPropertiesSet();
		return template;
	}

	@Bean(name = "stringsRedisTemplate")
	public StringRedisTemplate stringRedisTemplate() {
		StringRedisTemplate template = new StringRedisTemplate(connectionFactory());
		setMsgSerializer(template);
		template.setEnableTransactionSupport(false);
		template.afterPropertiesSet();
		return template;
	}
	
	@Override
	public CacheErrorHandler errorHandler() {
		return null;
	}
	
	public String getHost() {
		return host;
	}

	public void setHost(String host) {
		this.host = host;
	}

	public Integer getPort() {
		return port;
	}

	public void setPort(Integer port) {
		this.port = port;
	}

	public Integer getMaxActive() {
		return maxActive;
	}

	public void setMaxActive(Integer maxActive) {
		this.maxActive = maxActive;
	}
}
